package panda.query.tree;

import java.util.List;

import panda.query.scan.MergeJoinScanner;
import panda.query.scan.Scanner;
import panda.query.scan.UpdatableScanner;
import panda.query.struct.Attribute;
import panda.record.Schema;
import panda.transaction.Transaction;

/**
 * A class that performs a merge join operation on two children nodes
 * @author Tian Yuan
 */
public class MergeJoinNode implements QueryTreeNode {
	private Schema schema = new Schema();
	private QueryTreeNode l, r;
	private List<Attribute> la, ra;

	public MergeJoinNode(QueryTreeNode l, QueryTreeNode r, List<Attribute> la, List<Attribute> ra, Transaction tx) {
		// TODO check whether la and lb equal
		this.la = la;
		this.ra = ra;
		this.l = new SortNode(l, la, tx);
		this.r = new SortNode(r, ra, tx);
		schema.addAll(l.getSchema());
		schema.addAll(r.getSchema());
	}
	
	public Scanner open() {
		return new MergeJoinScanner((UpdatableScanner)l.open(), (UpdatableScanner)r.open(), la, ra);
	}
	
	public int blockAccessed() {
		return l.blockAccessed() + r.blockAccessed();
	}

	public int distinctSize(Attribute attr) {
		if (l.getSchema().hasAttribute(attr))
			return l.distinctSize(attr);
		else
			return r.distinctSize(attr);
	}


	public int outputSize() {
		double temp = 1.0;
		
		for (int i = 0; i < la.size(); i++)
			temp *= Math.max(l.distinctSize(la.get(i)), r.distinctSize(ra.get(i)));
		return (int)(Math.ceil((l.outputSize() * r.outputSize() * temp)));
	}
	
	public Schema getSchema() {
		return schema;
	}

}
